package org.vidi.amqp.basic;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.rabbitmq.client.Connection;
import org.vidi.amqp.common.ConnectionUtils;
import org.vidi.amqp.common.Constants;

import java.io.IOException;
import java.time.Duration;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.TimeoutException;
import java.util.function.BooleanSupplier;

/**
 * @author vidi
 */
public class ProducerConfirmAsync {

    public static void main(String[] args) {
        try (Connection connection = ConnectionUtils.newSingleConnection();
             Channel channel = connection.createChannel()) {
            channel.queueDeclare(Constants.PRODUCER_ASYNC_QUEUE_NAME, false, false, false, null);
            channel.confirmSelect();
            ConcurrentNavigableMap<Long, String> outStandingConfirm = new ConcurrentSkipListMap<>();
            ConfirmCallback cleanOutstandingConfirms = (sequenceNumber, multiple) -> {
                if (multiple) {
                    ConcurrentNavigableMap<Long, String> confirmed = outStandingConfirm.headMap(sequenceNumber, true);
                    confirmed.clear();
                } else {
                    outStandingConfirm.remove(sequenceNumber);
                }
            };

            channel.addConfirmListener(cleanOutstandingConfirms, (sequenceNumber, multiple) -> {
                String body = outStandingConfirm.get(sequenceNumber);
                System.err.format("Message with body %s has been nack-ed. Sequence number: %d, multiple: %b%n",
                        body, sequenceNumber, multiple);
                cleanOutstandingConfirms.handle(sequenceNumber, multiple);
            });

            for (int i = 0; i < Constants.MESSAGE_COUNT; i++) {
                String message = "PublishConfirmAsync-" + i;
                outStandingConfirm.put(channel.getNextPublishSeqNo(), message);
                channel.basicPublish("", Constants.PRODUCER_ASYNC_QUEUE_NAME, null, message.getBytes());

                if (!waitUntil(Duration.ofSeconds(60), outStandingConfirm::isEmpty)) {
                    throw new IllegalStateException("All messages could not be confirmed in 60 seconds");
                }
                System.out.format("Published %d messages and handled confirms asynchronously\n", i);
            }

        } catch (TimeoutException | IOException | InterruptedException e) {
            e.printStackTrace();
        }
    }

    private static boolean waitUntil(Duration timeout, BooleanSupplier condition) throws InterruptedException {
        int waited = 0;
        while (!condition.getAsBoolean() && waited < timeout.toMillis()) {
            Thread.sleep(10L);
        }
        return condition.getAsBoolean();
    }
}
